import json
import random
import traceback
import types

from re_common.baselibrary.tools.all_requests.aiohttp_request import AioHttpRequest
from re_common.baselibrary.tools.all_requests.mrequest import MRequest
from re_common.baselibrary.utils.basetime import BaseTime
from re_common.baselibrary.utils.core.mdeprecated import try_except2_async, try_except2, retry_func_async
from re_common.baselibrary.utils.core.requests_core import INSIDE_HEADERS

from apps.allsubdb.proxies_control.models import PostProxy
from apps.crawler_platform.core_abs.downstepabs import PareAdapter, deal_format, deal_load, BaseCoreABS
from apps.crawler_platform.core_callback import *
from apps.crawler_platform.core_callback.err_callback import all_client_back_callback
from apps.crawler_platform.core_platform.core_g import RedisKey
from apps.crawler_platform.core_platform.core_nosql import NoSqlUtil
from apps.crawler_platform.core_platform.core_sql import CoreSqlMixin
from apps.crawler_platform.core_platform.g_model import InputPlatformModel, TaskAllModel, \
    TaskInfoModel, TaskInfoSaveSettingModel, RedisAllTaskModel
from apps.crawler_platform.util.requestapihelper import RequestApiHelper
from settings import get_settings, URLDISTRIBUTED

STATICA_HACK = False
# globals()['kcah_acitats'[::-1].upper()] = False
if STATICA_HACK:  # pragma: no cover
    null()


class BaseCoreStep(BaseCoreABS):
    """
    实现层的底层，继承自DownStep抽象层
    """

    def __init__(self):
        super().__init__()
        self.pare_adapter = PareAdapter()
        self.err_tag = ""

        # ZCN:为保持结构一致性, 目前需要batch字段
        self.basetime = BaseTime()
        self.batchtime = self.basetime.get_beijin_date_strins()
        self.batch = self.batchtime[0:8] + "_" + self.batchtime[-6:]

    async def redis_callback(self, tmpdict: RedisAllTaskModel) -> RedisAllTaskModel:
        name = "_".join([self.pm.task_name, self.pm.task_tag, "redis_callback"])
        if name in globals():
            callback = eval(name)
            return await callback(tmpdict)
        else:
            return tmpdict

    @try_except2_async(callback=all_client_back_callback, is_print=True)
    async def get_redis(self):
        """
        获取redis的缓存
        """
        # 获取redis缓存的配置
        keys = self.pm.task_name + "_" + self.pm.task_tag
        # 获取redis setting的配置
        task_set = await self.pm.redis.hget(RedisKey.taskinfo_save_setting, keys)
        if not task_set:
            raise Exception("没有配置" + keys + "在" + RedisKey.taskinfo_save_setting + "表。")
        task_save_setting = TaskInfoSaveSettingModel.parse_raw(task_set)
        self.pm.task_setting = task_save_setting
        tmpdict = {}
        keys += "_"
        async for name, val in self.pm.redis.hscan_iter(RedisKey.taskinfo, match='{}*'.format(keys)):
            # tmp_key为1_1 即 数据库taskinfo中的groups_ordernum
            tmp_key = name.replace(keys, "")
            tmpdict[tmp_key] = TaskAllModel(
                task_set=task_save_setting,
                task_info=TaskInfoModel.parse_raw(val)
            )
        self.redis_model = RedisAllTaskModel(parse_dict=tmpdict)
        self.redis_model = await self.redis_callback(self.redis_model)
        self.pm.redis_model = self.redis_model
        # 获取redis缓存的配置
        keys = self.pm.task_name + "_" + self.pm.task_tag + "_cookie"
        # cookie
        cookie = await self.pm.redis.hget(RedisKey.cookie, keys)
        if cookie:
            self.pm.cookies = json.loads(cookie)
        return True, {}

    @try_except2_async(callback=all_client_back_callback, is_print=True)
    @retry_func_async(retry_times=3, sleep_time=1)
    async def get_proxy(self, task_name, task_tag):
        """
        获取代理接口
        """
        url = await URLDISTRIBUTED.get_redis_proxy_one()
        if self.input.input_code.strip() != "":
            if "10000" in self.input.input_code.split(";"):
                url = await URLDISTRIBUTED.get_redis_proxy_one()
            elif "10001" in self.input.input_code.split(";"):
                url = await URLDISTRIBUTED.get_qingting_proxy_one()
            elif "10003" in self.input.input_code.split(";"):
                url = await URLDISTRIBUTED.get_school_proxy_one()
            elif "10004" in self.input.input_code.split(";"):
                url = await URLDISTRIBUTED.get_zdaye_proxy_one()
            elif "10002" in self.input.input_code.split(";"):
                self.pm.proxy = random.choice(self.input.input_msg.split(";"))
                self.add_process_info("使用了传上来的代理")
                return True, {}
        elif self.pm.task_setting.proxy_type in ("qingting", "redis_taiyang"):
            if self.pm.task_setting.proxy_type == "qingting":
                url = await URLDISTRIBUTED.get_qingting_proxy_one()
            elif self.pm.task_setting.proxy_type == "redis_taiyang":
                url = await URLDISTRIBUTED.get_redis_proxy_one()
        else:
            data = InputPlatformModel(
                data=PostProxy(task_name=task_name,
                               task_tag=task_tag
                               )).json()
            rrq = RequestApiHelper.get_rrq()
            kwargs = {
                "rrq": rrq,
                "header": INSIDE_HEADERS,
                "url": get_settings().PROXY_URL,
                "timeout": 30,
                "data": data,
                "moths": MRequest.POST,
                "middler_list": [rrq.status_code_middlerwares,
                                 rrq.is_none_html_middlerwares,
                                 rrq.is_null_html_middlerwares,
                                 rrq.msg_status_code_middlerwares]
            }
            bools, status_msg_dict = await RequestApiHelper.proxy_api(**kwargs)
            if bools:
                self.pm.proxy = json.loads(rrq.html)["data"]
                if self.pm.proxy == "":
                    bools = False
            if not bools:
                self.add_process_info(self.step_info)
            return bools, status_msg_dict
        rrq = RequestApiHelper.get_rrq()
        kwargs = {
            "rrq": rrq,
            "header": INSIDE_HEADERS,
            "url": url,
            "timeout": 30,
            "moths": MRequest.GET,
            "middler_list": [rrq.status_code_middlerwares,
                             rrq.is_none_html_middlerwares,
                             rrq.is_null_html_middlerwares,
                             rrq.msg_status_code_middlerwares]
        }
        bools, status_msg_dict = await RequestApiHelper.proxy_api(**kwargs)
        if bools:
            self.pm.proxy = json.loads(rrq.html)["data"]["proxy"]
            if self.pm.proxy == "":
                bools = False
        if not bools:
            self.add_process_info(self.step_info)
        return bools, status_msg_dict

    @retry_func_async(retry_times=3, sleep_time=1)
    @try_except2_async(callback=all_client_back_callback, is_print=True)
    async def down(self, keys):
        """
        基础下载逻辑 20220622 认为代码无用 屏蔽
        :param keys:
        :return:
        """
        pass
        # self.step_info = self.add_all_process_info(f"获取代理1,key is {keys};")
        # bools, err_info = await self.get_proxy(self.pm.task_name, self.pm.task_tag)
        # if not bools:
        #     return bools, err_info
        # self.step_info = self.add_all_process_info(f"下载网页1,key is {keys};")
        # bools, err_info = await self.downler()
        # return bools, err_info

    def func_add(self, rrq=None):
        """
        rrq的方法添加器,通过callbakck的_m_list后缀添加
        :return:
        """
        if not rrq:
            rrq = self.rrq
        # 添加验证器
        name = "_".join([self.pm.task_name, self.pm.task_tag, "m", "list"])
        if name in globals():
            lists_m = eval(name)
            for item in lists_m:
                if item in self.pm.funcs:
                    func_m = types.MethodType(eval(item), rrq)
                    rrq.__dict__.update({item: func_m})

    @try_except2_async(callback=all_client_back_callback, is_print=True)
    async def downler(self):
        """
        具体的下载逻辑,核心的下载器 20220622 认为代码无用 屏蔽
        """
        pass
        # self.rrq = AioHttpRequest()
        # # 添加验证器
        # self.func_add()
        # Lists_middler = []
        # self.rrq.set_url(self.pm.url) \
        #     .set_timeout(self.pm.time_out) \
        #     .set_header(self.pm.this_header)
        # if self.pm.proxy:
        #     self.rrq.set_proxy(self.pm.proxy)
        #
        # if self.pm.resp_middler:
        #     self.rrq.set_middler_para(self.pm.resp_middler)
        #     for k, v in self.pm.resp_middler.items():
        #         Lists_middler.append(eval("self.rrq." + k))
        #
        # if self.pm.marks:
        #     self.rrq.set_marks(self.pm.marks)
        #     Lists_middler.append(self.rrq.marks_middlerwares)
        # self.rrq.set_middler_list(Lists_middler)
        # if self.pm.params:
        #     self.rrq.set_params(self.pm.params)
        # if self.pm.req_type == "POST":
        #     if self.pm.post_data:
        #         self.rrq.set_data(self.pm.post_data)
        #     bools, dicts = await self.rrq.run(MRequest.POST)
        # elif self.pm.req_type == "GET":
        #     bools, dicts = await self.rrq.run(MRequest.GET)
        # else:
        #     raise Exception("req_type 错误;" + self.pm.req_type)
        # return bools, dicts

    @try_except2()
    def para_redis(self, keys):
        """
        解析redis的缓存得到各种变量
        """
        # redis_model 中存储着所有key的配置
        self.pm.redis_model = self.redis_model
        self.pm.one_para = self.redis_model.parse_dict[keys]
        self.pm.task_tag_next = self.pm.one_para.task_info.task_tag_next
        self.pm.url = self.pm.one_para.task_info.url
        bools, msg = deal_format(self.pm.url, self.pm.sql_model.get_dict())
        if not bools:
            return bools, msg
        else:
            self.pm.url = msg
        self.pm.req_type = self.pm.one_para.task_info.req_type
        self.pm.this_header = deal_load(self.pm.one_para.task_info.headers)
        for k, v in self.pm.this_header.items():
            br, msge = deal_format(v, self.pm.sql_model.get_dict())
            if not br:
                return br, msge
            else:
                v = msge
                self.pm.this_header[k] = v
        bools_marks, msg_marks = deal_format(self.pm.one_para.task_info.marks, self.pm.sql_model.get_dict())
        if not bools_marks:
            return bools, msg_marks
        else:
            marks_t = msg_marks
        self.pm.set_marks(marks_t)
        self.pm.resp_middler = deal_load(self.pm.one_para.task_info.resp_middler)
        self.pm.time_out = self.pm.one_para.task_info.timeout
        self.pm.validate_rules = ""
        if len(self.pm.one_para.task_info.validate_rules.strip()) > 0:
            self.pm.validate_rules = deal_load(self.pm.one_para.task_info.validate_rules)
        self.pm.order_num = self.pm.one_para.task_info.order_num
        self.pm.groups = self.pm.one_para.task_info.groups
        self.pm.funcs = self.pm.one_para.task_info.funcs
        self.pm.turn_page = self.pm.one_para.task_info.turn_page
        self.pm.proxy_state = self.pm.one_para.task_info.proxy_state
        self.pm.set_state_key(self.pm.one_para.task_info.state_key)
        self.pm.post_data_is_json = self.pm.one_para.task_info.post_data_is_json
        self.pm.others_parameter = deal_load(self.pm.one_para.task_info.others_parameter)
        self.pm.resp_encoding = self.pm.others_parameter.get("resp_encoding", None)
        self.pm.resp_errors = self.pm.others_parameter.get("resp_errors", "strict")
        self.pm.allow_resp_text = eval(self.pm.others_parameter.get("allow_resp_text", "True"))
        self.pm.allow_resp_bytes = eval(self.pm.others_parameter.get("allow_resp_bytes", "False"))
        self.pm.is_grpc = eval(self.pm.others_parameter.get("is_grpc", "False"))
        self.pm.type_request = self.pm.others_parameter.get("type_request", "aiohttp")
        self.pm.force_close = self.pm.others_parameter.get("force_close", None)
        self.pm.set_cookies = eval(self.pm.others_parameter.get("set_cookies", "False"))
        self.pm.get_cookies = eval(self.pm.others_parameter.get("get_cookies", "False"))
        self.pm.set_post_data()
        self.pm.set_params()
        return True, ""

    @try_except2_async(callback=all_client_back_callback, is_print=True)
    async def para_next(self):
        """
        解析得出下一级需要的数据
        """
        bools, dicts = await self.pare_adapter.pare_htmlmodel(self.pm.down_model, self.redis_model)
        if self.pare_adapter.para_dicts is not None:
            self.para_dicts = self.pare_adapter.para_dicts
        return bools, dicts

    @retry_func_async(retry_times=3, sleep_time=2)
    @try_except2_async(callback=all_client_back_callback, is_print=True)
    async def save_nosql_html(self):
        """
        存储到nosql数据库
        """
        # 期刊列表级默认保存历史
        self.pm.set_para()
        return await NoSqlUtil.save_nosql_html(self.pm)

    @retry_func_async(retry_times=3, sleep_time=2)
    @try_except2_async(callback=all_client_back_callback, is_print=True)
    async def save_nosql_html_etl(self):
        """
        ETL存储到nosql数据库
        """
        return await NoSqlUtil.save_nosql_html_etl(self.pm)

    @retry_func_async(retry_times=3, sleep_time=2)
    @try_except2_async(callback=all_client_back_callback, is_print=True)
    async def read_nosql_html(self):
        """
        存储到nosql数据库
        """
        # 期刊列表级默认保存历史
        return await NoSqlUtil.read_nosql_html(self.pm)

    # @retry_func_async(retry_times=3, sleep_time=2)
    # @try_except2_async(callback=all_client_back_callback, is_print=True)
    # async def read_nosql_latest(self):
    #     """
    #     读取mongo里latest表内容
    #     """
    #     return await NoSqlUtil.read_nosql_latest(self.pm)

    @try_except2_async(callback=all_client_back_callback, is_print=True)
    async def err_collect(self, err_info, pm, batch, tag):
        if isinstance(err_info, dict):
            v = {"err_msg": str(err_info)}
            # 不在写入sql 202305029
            # await CoreSqlMixin.insert_errinfo(pm.task_name, pm.task_tag, pm.Id, batch, v["err_msg"],
            #                                   tag)
            err_info["id"] = pm.Id
            # err_info["err_msg"] = v["err_msg"]
            return True, err_info

        else:
            return False, err_info

    async def save_sql(self, table, next_table, type="", msg=""):
        """

        :param table:
        :param type:
        :param code: 这里的code不只是起状态码的作用，可以是其他作用，不要被命名误导
        :return:
        """
        where = self.pm.get_pkm()
        vv = self.pm.get_sql_state()
        # 中途失败的状态码更新到sql，这里要考虑外围save_sql无法调用的情况，比如步骤没有走到save_sql那一步
        if type == "save_state":
            # # 这里的code 相当于errmsg
            # self.pm.set_sql_state({"err_msg": str(msg)})
            # # 获取stepinfo的state_key的字典
            #
            # state_key_dict = self.pm.get_state_key(type="eval")
            # if not state_key_dict:
            #     # 如果不存在，可能是没有走到for循环，也可能是redis都没有加载进来，频繁出现这种情况应该报警，
            #     # 但目前没有想到相关逻辑怎么做,需要与错误提醒配合
            #     pass
            # if msg is not None and isinstance(msg, dict):
            #     if msg.get("code", 0) != 0 or msg.get("msg_code", 0) != 0:
            #         # 获取对应的状态key
            #         sql_key = state_key_dict["s_key"]
            #         # 设置状态码
            #         if msg.get("code", 0) != 0:
            #             self.pm.set_sql_state({sql_key: int(msg["code"])})
            #         else:
            #             self.pm.set_sql_state({sql_key: int(msg["msg_code"])})
            # # 这里是失败后的设置状态，最终状态不能被写1，也有可能是其他，但暂时无法区分哪些状态应该允许，哪些状态不被允许
            # # 所以这里统一写为0，不会影响逻辑，这里的last_key可能和s_key是一样的，可能需要逻辑处理
            # last_key = state_key_dict["last_key"]
            # if last_key in vv.keys():
            #     vv[last_key] = 0
            # # 获取对应的状态key
            # f_key = state_key_dict["f_key"]
            # f_last_key = state_key_dict.get("f_last_key", None)
            # update_no_placeholder_temp = {}
            # histroy_sql_set = self.pm.get_histroy_sql_set()
            # if f_key and f_key not in histroy_sql_set.keys():
            #     update_no_placeholder_temp.update({f_key: f_key + "+1"})
            # if f_last_key and f_last_key not in histroy_sql_set.keys():
            #     update_no_placeholder_temp.update({f_last_key: f_last_key + "+1"})
            # self.pm.set_update_no_placeholder(update_no_placeholder_temp)
            # update_no_placeholder = self.pm.get_update_no_placeholder()
            # if vv or update_no_placeholder:
            #     bools, dicts = await CoreSqlMixin.update_sql(table, vv, update_no_placeholder, where)
            #     if bools:
            #         # 如果失败加1的逻辑执行成功，就记录在内存，防止多次加1，因为一个rawid的请求有可能由多条url组成，而每个url
            #         # 都会调用这个+1 逻辑
            #         self.pm.set_histroy_sql_set(update_no_placeholder_temp)
            #     return bools, dicts
            # else:
            #     return True, {}

            self.statemanager.set_state(self.pm)
            where = self.pm.get_pkm()
            self.pm.set_sql_state({"err_msg": str(msg)})
            vv.update(self.pm.get_sql_state())
            # self.statemanager.debug_dicts(self.pm)
            self.pm.set_sql_state(vv)
            #####################
            update_no_placeholder = self.pm.get_update_no_placeholder()
            if vv or update_no_placeholder:
                bools, dicts = await CoreSqlMixin.update_sql(table, vv, update_no_placeholder, where)
                return bools, dicts
            else:
                return True, {"msg": "没有执行最后的更新sql"}
        # 这里的作用是设置失败加1
        elif type == "set_failcount":
            # 设置失败加一策略
            update_no_placeholder = {}
            # code 代表键
            histroy_sql_set = self.pm.get_histroy_sql_set()
            # 获取过程所有状态
            state_key_dict = self.pm.get_state_key(type="eval")
            for keys in msg.split(";"):
                # code 传入的键在里面的话
                if keys in state_key_dict.keys():
                    # 获取对应的值，这个值是数据库的键
                    fail_key = state_key_dict[keys]
                    ###########################
                    # 不存在就放进去执行，防止失败加1重复执行几次
                    if fail_key not in histroy_sql_set.keys():
                        update_no_placeholder[fail_key] = fail_key + "+1"
            # 放在这里，不放入pm污染pm
            if vv or update_no_placeholder:
                bools, dicts = await CoreSqlMixin.update_sql(table, vv, update_no_placeholder, where)
                if bools:
                    # 如果失败加1的逻辑执行成功，就记录在内存，防止多次加1，因为一个rawid的请求有可能由多条url组成，而每个url
                    # 都会调用这个+1 逻辑
                    self.pm.set_histroy_sql_set(update_no_placeholder)
                return bools, dicts
            else:
                return True, {}

        elif type == "set_failcount_first":
            # 为什么要这个 因为在for循环中 对配置 的eval有可能报错而导致无法对失败加1，这里提前处理
            # 为什么要try 不想这里里面的错误影响后面
            try:
                # 设置失败加一策略
                update_no_placeholder = {}
                # code 代表键
                histroy_sql_set = self.pm.get_histroy_sql_set()
                # 获取过程所有状态
                for k, v in self.pm.redis_model.parse_dict.items():
                    state_key_dict = eval(self.pm.redis_model.parse_dict[k].task_info.state_key)
                    for keys in msg.split(";"):
                        # code 传入的键在里面的话
                        if keys in state_key_dict.keys():
                            # 获取对应的值，这个值是数据库的键
                            fail_key = state_key_dict[keys]
                            ###########################
                            # 不存在就放进去执行，防止失败加1重复执行几次
                            if fail_key not in histroy_sql_set.keys():
                                update_no_placeholder[fail_key] = fail_key + "+1"
                # 放在这里，不放入pm污染pm
                if update_no_placeholder:
                    bools, dicts = await CoreSqlMixin.update_sql(table, {}, update_no_placeholder, where)
                    if bools:
                        # 如果失败加1的逻辑执行成功，就记录在内存，防止多次加1，因为一个rawid的请求有可能由多条url组成，而每个url
                        # 都会调用这个+1 逻辑
                        self.pm.set_histroy_sql_set(update_no_placeholder)
                    return bools, dicts
                else:
                    return True, {}
            except:
                # TODO 后面收集这里的错误
                traceback.print_exc()
                # 这里不影响后面的流程，所以返回True
                return True, {}
        else:
            raise Exception("没有type的匹配项")

    @retry_func_async(retry_times=3, sleep_time=2)
    @try_except2_async(callback=all_client_back_callback, is_print=True)
    async def do_etl_remote(self, inputs):
        url = await URLDISTRIBUTED.get_etl_local_do_batch_url()
        data = inputs.json()
        headers = INSIDE_HEADERS
        rrq = RequestApiHelper.get_rrq()
        kwargs = {
            "rrq": rrq,
            "header": headers,
            "url": url,
            "timeout": 120,
            "data": data,
            "moths": MRequest.POST,
            "middler_list": [rrq.status_code_middlerwares,
                             rrq.msg_status_code_middlerwares]
        }
        bools, dicts = await RequestApiHelper.etl_remote_meta(**kwargs)
        self.pm.resp_html = rrq.html
        return bools, dicts

    @retry_func_async(retry_times=3, sleep_time=2)
    @try_except2_async(callback=all_client_back_callback, is_print=True)
    async def do_etl_remote_ref(self, inputs):
        url = await URLDISTRIBUTED.get_etl_local_do_batch_ref_url()
        data = inputs.json()
        headers = INSIDE_HEADERS
        rrq = RequestApiHelper.get_rrq()
        kwargs = {
            "rrq": rrq,
            "header": headers,
            "url": url,
            "timeout": 120,
            "data": data,
            "moths": MRequest.POST,
            "middler_list": [rrq.status_code_middlerwares,
                             rrq.msg_status_code_middlerwares]
        }
        bools, dicts = await RequestApiHelper.etl_remote_meta(**kwargs)
        self.pm.resp_html = rrq.html
        return bools, dicts

    async def my_task(self, table=""):
        """
        接入调用的task
        :return:
        """
        try:
            bools, err_info = await self.run_task()
            if self.pm.time_use.all_time > 100:
                await self.pm.time_use.write_nosql()

            if not bools:
                bools_err, err_info_after = await self.err_collect(err_info, self.pm, self.batch, self.err_tag)
                if get_settings().GET_INFO_DEBUG == "true":
                    err_info["all_process_info"] = self.all_process_info
                if err_info:
                    err_info["process_info"] = self.process_info
                # 保底打标
                vv = self.statemanager.deal_my_task_failed(bools, err_info)
                if vv:
                    vv.update({"err_msg": str(err_info)})
                    where = self.pm.get_pkm()
                    await CoreSqlMixin.update_sql(table, vv, {}, where)
                # ZCN: 临时使用, 为了处理在下载之外的解析之类的内部错误, 即使有错, 也会将这个标志置为 1, 因此在此对有错误且标志为1的数据进行
                # 判定, 如果为1 则将其打为 909, 以免干扰正常数据
                # state_dict = self.pm.get_sql_state()
                # for k, v in state_dict.items():
                #     where = self.pm.get_pkm()
                #     if v == 1:
                #         await CoreSqlMixin.update_parse_wrong_info(table, k, where)
                return bools, err_info_after
            return bools, err_info

        except:
            # await CoreSqlMixin.insert_errinfo(self.pm.task_name, self.pm.task_tag, self.pm.Id, self.batch,
            #                                   str(traceback.format_exc()), self.err_tag)
            return False, {"traceback": traceback.format_exc()}

    def deal_html_etl(self, *args, **kwargs):
        """
        处理html
        :return:
        """
        return self.para_next(args, kwargs)

    def get_html(self, *args, **kwargs):
        """
        获取html数据，加载到model中
        """
        pass
